use futures_util::stream::StreamExt;
use rand::distributions::Alphanumeric;
use rand::Rng;

use prost::Message as ProstMessage;

use rdkafka::config::ClientConfig;
use rdkafka::consumer::stream_consumer::StreamConsumer;
use rdkafka::consumer::{BaseConsumer, CommitMode, Consumer};
use rdkafka::message::Message as KafkaMessage;

const SERVER: &'static str = "coin-kafka.corp.prestolabs.io:9092";
const LEVERAGE: f64 = 10.0;

fn get_topic_list() -> Vec<String> {
    let consumer: BaseConsumer = ClientConfig::new()
        .set("bootstrap.servers", SERVER)
        .create()
        .expect("Consumer creation failed");

    let metadata = consumer
        .fetch_metadata(None, std::time::Duration::from_secs(5))
        .expect("Failed to fetch metadata");

    let topics: Vec<String> = metadata
        .topics()
        .iter()
        .map(|topic| topic.name().to_string())
        .filter(|s| s.starts_with("STRAT_"))
        .filter(|s| s.contains("lab"))
        .collect();
    topics
}

struct Info {
    currency: String,
    balance: f64,
    required_margin: f64,
}

async fn read_balance(topic: &str) -> anyhow::Result<Info> {
    // "STRAT_strategy-50.ap-northeast-1.huobi_lm_agg2_huobi_futures_xrp_lab"
    // "STRAT_strategy-69.aliyun-cn-hongkong.okex_lm_agg2_okex_futures_xrp_usdt_lab"
    let re = regex::Regex::new("^STRAT_(.+)_lm_agg2_(.+)_futures_([a-z]+)_(usdt_)?lab$").unwrap();
    let caps = re
        .captures(topic)
        .ok_or(anyhow::format_err!("invalid topic pattern: {}", topic))?;
    let _machine = &caps[1];
    let exchange = &caps[2];
    let base = (&caps[3]).to_uppercase();
    let quote = if topic.contains("usdt") {
        "USDT"
    } else {
        "USD"
    };

    // BTC-USD => 1 position = 100 USD
    // others-USD => 1 position = 10 USD
    let multiplier: f64 = if quote == "USD" {
        if base == "BTC" {
            100.0
        } else {
            10.0
        }
    } else {
        match base.as_ref() {
            "BTC" => 0.01,
            "EOS" => 10.0,
            "ETH" => 0.1,
            "LTC" => 1.0,
            "BCHABC" => 0.1,
            "XRP" => 100.0,
            "ETC" => 10.0,
            "BCHSV" => 1.0,
            _ => unreachable!(),
        }
    };

    let mut required_margin_total: f64 = 0.0;

    let path = format!(
        "../../../coin_deploy/lm_agg2/config_coin1_style/lab/futures_{}/{}-{}.json",
        exchange, base, quote
    );
    let f = std::fs::read_to_string(path)?;
    let j: serde_json::Value = serde_json::from_str(&f)?;
    for (_symbol, dic) in j["products"].as_object().unwrap() {
        let max_pos = dic["max_pos"].as_f64().unwrap();
        let min_pos = dic["min_pos"].as_f64().unwrap();

        let required_margin = max_pos.max(-min_pos) * multiplier / LEVERAGE;
        required_margin_total += required_margin;
    }

    let group_id: String = rand::thread_rng()
        .sample_iter(Alphanumeric)
        .take(10)
        .collect();

    let consumer: StreamConsumer = ClientConfig::new()
        .set("bootstrap.servers", SERVER)
        .set("group.id", &group_id)
        .create()
        .expect("Consumer creation failed");

    consumer.subscribe(&[topic]).expect("Can't subscribe");
    let mut message_stream = consumer.start();

    while let Some(message) = message_stream.next().await {
        match message {
            Err(e) => anyhow::bail!("Kafka error: {}", e),
            Ok(m) => {
                if let Some(bytes) = m.payload() {
                    let log = proto::coin::proto::StrategyLog::decode(bytes)
                        .expect("not StrategyLog proto");

                    if log.r#type
                        == Some(proto::coin::proto::strategy_log::LogType::PnlBalance as i32)
                    {
                        if let Some(pnl_balance) = log.pnl_balance {
                            if let Some(currency) = pnl_balance.currency {
                                if let Some(total) = pnl_balance.total {
                                    return Ok(Info {
                                        currency,
                                        balance: total,
                                        required_margin: required_margin_total,
                                    });
                                }
                            }
                        }
                    }
                }
                consumer.commit_message(&m, CommitMode::Async).unwrap();
            }
        };
    }
    unreachable!();
}

#[tokio::main]
async fn main() {
    env_logger::init();

    let topics = get_topic_list();
    let bars = indicatif::MultiProgress::new();
    let mut handles = vec![];

    for topic in topics {
        let pb = indicatif::ProgressBar::new_spinner();
        bars.add(pb.clone());

        pb.enable_steady_tick(1000);
        pb.set_message(&topic);

        handles.push(tokio::spawn(async move {
            match tokio::time::timeout(std::time::Duration::from_secs(30), read_balance(&topic))
                .await
            {
                Ok(result) => match result {
                    Ok(info) => {
                        let ratio = info.balance / info.required_margin * 100.0;
                        let msg = format!(
                            "{:100} | {:6} - {:10.2} (current) / {:10.2} (required) = {:7.2} %",
                            topic, info.currency, info.balance, info.required_margin, ratio,
                        );
                        let msg = format!(
                            "{}",
                            if ratio > 300.0 {
                                console::style(msg).green()
                            // msg.green()
                            } else if ratio < 100.0 {
                                console::style(msg).red()
                            // msg.red()
                            } else {
                                console::style(msg)
                                // msg.normal()
                            }
                        );
                        pb.set_message(&msg);
                    }
                    Err(e) => {
                        pb.set_message(&e.to_string());
                    }
                },
                Err(_) => {
                    let msg = &format!("{:100} | [Timeout]", topic);
                    let msg = format!("{}", console::style(msg).red());
                    pb.set_message(&msg);
                }
            }
            pb.finish();
        }));
    }
    bars.join().unwrap();
    futures::future::join_all(handles).await;
}
